# -*- coding: utf-8 -*-
# standard
import os
import sys
import multiprocessing
from time import sleep
from threading import Thread
from queue import Queue
from queue import Empty
from concurrent.futures import ThreadPoolExecutor
import threading

# third
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from watchdog.events import FileDeletedEvent
from watchdog.events import FileCreatedEvent
from watchdog.events import FileModifiedEvent


# local
_P_PATH =  os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
if _P_PATH not in sys.path:
    sys.path.append(_P_PATH)
from models import *


"""
@Title:  生产环境的函数 
@File: pro_funcs.py
@Author: walle 2023年10日20日 09时31分00秒
@Version: 1.0.0
@Desc: 
"""


DOWNLOAD_TASKS = Queue()  # 下载任务 元素是 DownloadTask 对象
COPY_TASKS = Queue()  # 进程拷贝任务 元素是 ScpTask 对象
# THREAD_COPY_TASKS = Queue()  # 线程拷贝任务 元素是 ScpTask 对象
THREADS = None  #  线程池
FLAG = False
COND = threading.Condition()  # google服务器的ban事件
# BAN.acquire()


def set_ban(is_ban: bool):
    # 一个用于测试的辅助函数。用于调整 BAN 标志位
    global FLAG
    FLAG = is_ban
    logger.debug(f"BAN 状态已设置为 {FLAG}")


def get_ban_status():
    return FLAG



class DownloadTask(BaseModel):
    # 下载任务， 放入线程执行比较好
    id: int = Field(..., description="数据库中对应记录的id")
    num: int = Field(..., description="序号")
    key: str = Field(..., description="视频的key")
    url: str = Field(..., description="视频对应的页面地址")
    is_downloaded: int = Field(..., description="是否已下载？1已下载，-1 出错, 0 没下载")
    
    def execute_download(self):
        # 执行下载
        file = VideoFile(download_url=self.url, key=self.key)
        resp = Resp()
        try:
            resp = file.download(ban_cb=set_ban, re_download=False)
        except Exception:
            logger.error(traceback.format_exc())
            resp.message = "执行下载任务失败"
        if resp:
            logger.debug(f"第{self.num}个视频({self.key})的下载成功")
            file.save_orm(db_handler=new_db_session())     
            scp_task = ScpTask(num=self.num, key=self.key, sources=file.save_paths)     
            COPY_TASKS.put(scp_task)  # 放入进程队列
            logger.debug(f"{file.key}已完成下载环节")
        else:
            logger.error(f"第{self.num}个视频({self.key})的下载失败, 原因: {resp.message}")
        try:
            DOWNLOAD_TASKS.task_done()
        except Exception as e:
            print(e)
        return resp


def copy_by_process(need_scp_tasks: multiprocessing.Queue, thread_num: int = 1):
    """
    使用进程拷贝文件
    """
    logger.debug(f"本地scp队列长度:{COPY_TASKS.qsize()}")
    # 先启动拷贝线程
    for i in range(thread_num):
        t = Thread(target=ScpRecordOrm.copy_to_remote_forever, kwargs={"tasks": COPY_TASKS}, name=f"scp_thread_{i + 1}", daemon=True)
        t.start()


class MyHandler(FileSystemEventHandler):  
    """
    文件监视器
    :param FileSystemEventHandler: _description_
    """
    def on_modified(self, event: FileModifiedEvent):   
        # 文件被修改事件
        file_path = event.src_path
        logger.debug(f"文件 {file_path} 已被修改")  
        if os.path.isdir(file_path):
            return  # 文件夹的改变不理会
        sleep(5)  # 等待5秒，防止文件没写完
        all_paths = [file_path]
        new_downloads = RawJsonHandler.read_json_data(all_paths)
        num = 0
        for new_download in new_downloads:
            num += 1
            DOWNLOAD_TASKS.put(DownloadTask(**new_download, num=num))

    def on_created(self, event: FileCreatedEvent): 
        # 新文件事件
        file_path = event.src_path
        logger.debug(f"文件 {file_path} 已被创建")  
        if os.path.isdir(file_path):
            return  # 文件夹的改变不理会
        sleep(5)  # 等待5秒，防止文件没写完
        all_paths = [file_path]
        new_downloads = RawJsonHandler.read_json_data(all_paths)
        num = 0
        for new_download in new_downloads:
            num += 1
            DOWNLOAD_TASKS.put(DownloadTask(**new_download, num=num))

    def on_deleted(self, event: FileDeletedEvent):  
        # 删除事件
        logger.debug(f"文件 {event.src_path} 已被删除")  


def read_json_forever():
    """
    一直读json 文件。用于调试
    """
    event_handler = MyHandler()  
    observer = Observer()  
    observer.schedule(event_handler, path=RAW_JSON_DIR, recursive=True) 
    observer.start()
    observer.join()


def start_watch_json_dir() -> Thread:
    """
    开始监控 json 文件夹
    """
    # 先读一次
    new_downloads = RawJsonHandler.read_json_data()
    num = 0
    for new_download in new_downloads:
        num += 1
        DOWNLOAD_TASKS.put(DownloadTask(**new_download, num=num))
    # 开始监控
    event_handler = MyHandler()  
    observer = Observer()  
    observer.schedule(event_handler, path=RAW_JSON_DIR, recursive=True) 
    observer.start()
    return observer


def supplement_raw_video_info(id: int, key: str):
    """
    把已经复制到scp服务器但是没有被正确标记的 RawVideoObjectOrm 修复。
    """
    db_session = new_db_session()
    one: RawVideoObjectOrm = db_session.query(RawVideoObjectOrm).where(RawVideoObjectOrm.id == id).one_or_none()
    if one is None:
        logger.error(f"没有在数据库找到{key}对应的记录，这可能需要手动修复")
    else:
        one.is_downloaded = 1
        one.download_time = datetime.now()
        db_session.commit()
        logger.warning(f"视频({key})已拷贝但是未被标记为已下载，修复")
    db_session.close()
    

def init_download_tasks(check_remote_files: bool = True):
    """
    初始化下载任务
    todo: 执行此任务之前，必须先把downloads下面的文件全部上传到scp服务器，以尽量避免重复下载
    :param check_remote_files: 是否检查远程文件完整性
    """
    if check_remote_files:
        good_files, _bad_files = ScpRecordOrm.view_remote_files()  # 远程服务器上已经scp的文件
        # _bad_files 可能是已下载未来得及拷贝的情况，不要处理仅仅提醒即可。ScpRecordOrm.view_remote_files 函数已经打印日志提醒过了
    else:  # 不检查远程文件
        good_files = []
    raw_need_download_videos = RawVideoObjectOrm.get_all_videos_from_db()  # 会查询没有下载，和下载时被ban的记录。
    # todo： 先取后10个做测试
    # need_download_videos = need_download_videos[len(need_download_videos) - 4:]  
    temp_data = {}
    if DOWNLOAD_TASKS.qsize() != 0:      
        while DOWNLOAD_TASKS.qsize() != 0:
            task = DOWNLOAD_TASKS.get()
            temp_data[task.key] = task.dict()
    for x in raw_need_download_videos:
        temp_data[x['key']] = x
    need_download_videos = list(temp_data.values())
    num = 0
    for x in need_download_videos:
        if x['key'] in good_files:   # 在scp服务器上有记录的文件，修复  状态不正确的修复状态
            logger.warning(f"视频{x['key']}的下载状态异常{x['is_downloaded']}, 进行修复")
            supplement_raw_video_info(x['id'], x['key'])  # 
        else:      
            num += 1  # 没有下载过和下载失败的 x['is_downloaded'] in [-2, 0]
            DOWNLOAD_TASKS.put(DownloadTask(**x, num=num))
    logger.debug(f"从数据库加载了{num}个待下载视频放入下载工作队列")



def init_scp_tasks():
    """
    初始化拷贝任务
    """
    ScpRecordOrm.check_downloaded_folder(delete_file=False)  # 检查已下载的视频的大小是否合法？未确认是否合法之前暂不执行删除。
    need_scp_tasks = ScpRecordOrm.get_need_copy_records()  # 取出已下载的视频的清单
    num = 0
    for x in need_scp_tasks:
        num += 1
        COPY_TASKS.put(ScpTask(num=num, key=x['key'], sources=x['sources']))
    logger.debug(f"从{DOWNLOAD_DIR}目录加载了 {num} 个待等待scp的视频放入工作队列")


def download_video_in_thread(tasks: Queue):
    """
    下载视频的线程函数
    """
    empty_flag = False
    min_timeout, max_timeout = 0.1, 30
    timeout = min_timeout # 检测ban的等待超时,一旦发生ban事件,这个值就要被修改成 max_timeout,ban状态解除时改为 min_timeout
    while 1:
        self = threading.current_thread()
        task = None
        try:
            task: DownloadTask = tasks.get(timeout=1)
        except Empty:
            if not empty_flag:
                empty_flag = True
                logger.debug(f"线程 {self.name} 已无可用任务")
        except Exception:
            logger.exception(f"线程 {self.name} 获取任务时发生错误")
        finally:
            if task:
                begin = datetime.now()
                while 1:                
                    COND.acquire()
                    try:
                        """
                        Condition 的 wait_for 用于等待 predicate 的条件为真
                        Condition 还有一个 wait 方法用来等待 notify，也有第二个参数 timeout
                        timeout 参数表示在 predicate 方法返回值为真之前，最多等待多少时间。
                        如果等待超时，返回 predicate 的结果
                        """
                        is_ban = COND.wait_for(predicate=get_ban_status, timeout=timeout)  
                        if is_ban:  # 仍然在ban状态中,
                            # logger.debug(f"检测到ban事件, 线程 {self.name} 休眠 {timeout} 秒")
                            if timeout != max_timeout:  # 之前是正常状态，现在转入 ban 状态
                                timeout = max_timeout
                                logger.debug(f"检测到ban状态, 线程 {self.name} 进入休眠状态")
                                continue
                            if datetime.now() - begin > timedelta(seconds=timeout):  # 等待到最大超时了，解除ban状态
                                logger.debug(f"等待到最大超时 {max_timeout}, 线程 {self.name} 解除ban状态")
                                if FLAG:
                                    set_ban(is_ban=False)
                                break  
                        else:
                            if timeout == max_timeout:  # 需要解除 ban 状态
                                timeout = min_timeout
                                end = datetime.now()
                                d = (end - begin).total_seconds()
                                logger.debug(f"检测到ban状态解除, 线程 {self.name} 恢复. 共计休眠 {d} 秒")
                            else:  # 这是没发生过 ban 事件或者 ban 状态已经解除过的状态。
                                logger.debug(f"没有检测到ban事件, 线程 {self.name} 正常执行")
                            break  # ban状态解除或者没发生ban,终止等待
                    except Exception as e:
                        logger.exception(e)
                    finally: 
                        COND.release()  #  这里保证了 条件对象一定会被释放
                # 继续执行
                sleep(randint(1, 3))  # 休眠 随机时间
                resp = task.execute_download()     
                if resp.message == DownloadResponseError.BAN.value:
                    tasks.put(task)
            else:
                pass


def start_download_service(thread_num: int = 0):
    """
    开始下载服务，本函数不会阻塞
    :param thread_num: _description_, defaults to 0
    """
    thread_num = 10 if thread_num == 0 else thread_num
    for i in range(thread_num):
        t = Thread(target=download_video_in_thread, kwargs={"tasks": DOWNLOAD_TASKS}, name=f"downloader_{str(i + 1).zfill(2)}", daemon=True)
        t.start()
    # 模拟BAN事件
    # set_ban(is_ban=True)
    # threading.Timer(interval=1, function=set_ban, kwargs={"is_ban": True}).start()
    # threading.Timer(interval=13, function=set_ban, kwargs={"is_ban": False}).start()



def foo():
    
    logger.info("程序启动完成")
    while 1:
        try:
            sleep(0.1)
        except KeyboardInterrupt:
            logger.info("程序已停止")
            break

        
def concurrency_run(check_remote_files: bool = True, download_workers: int = 1, scp_workers: int = 1):
    """
    并发模式执行任务
    :param check_remote_files: 是否检查远程文件完整性
    :param download_workers: 下载线程数
    :param scp_workers: scp 线程数

    """
    # 监控json文件夹的改变，这里不需要多线程。
    start_watch_json_dir()
    # 初始化待下载的视频队列
    init_download_tasks(check_remote_files=check_remote_files)
    # 初始化scp任务队列
    init_scp_tasks()
    # 启动下载服务
    start_download_service(thread_num=download_workers)


    # 启动拷贝线程 ，注意这个方法是非阻塞的
    copy_by_process(thread_num=scp_workers, need_scp_tasks=COPY_TASKS)
    # 阻塞线程
    foo()



if __name__ == '__main__':
    concurrency_run(check_remote_files=False)
    pass